package com.github.mq;

import cn.hutool.core.util.StrUtil;
import com.github.bean.ServerUser;
import com.github.constant.MyConstant;
import com.github.manager.Manager;
import com.github.util.AESEncrypt;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class StreamMessageListener implements StreamListener<String, MapRecord<String, String, String>> {
    @Autowired
    private RedissonClient redissonClient;
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        if (message == null) {
            log.info("接收到的消息为空");
            return;
        }

        try {
            threadPoolExecutor.execute(() -> processStreamMessage(message.getValue()));
        } catch (Exception e) {
            System.err.println("处理消息时发生错误: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private void processStreamMessage(Map<String, String> messageData) {

        String sessionId = messageData.get("sessionId");
        String deviceType = messageData.get("deviceType");
        String message = messageData.get("message");
        String account = messageData.get("account");
        if (StrUtil.isBlank(account) && StrUtil.isBlank(sessionId)) {
            log.info("accountId和sessionId同时为空");
            return;
        }
        ServerUser serverUser = null;
        if (StrUtil.isNotBlank(account)) {
            serverUser = Manager.accountUserMap.getOrDefault(account, null);
        } else if (StrUtil.isNotBlank(sessionId)) {
            serverUser = Manager.sessionIDUserMap.getOrDefault(sessionId, null);
        }

        if (serverUser == null) {
            log.info("serverUser不存在，发送消息失败");
            return;
        }
        WebSocketSession serverSession = null;
        if (deviceType.equals("1")){
            serverSession = serverUser.getServerSession();
        }else if (deviceType.equals("2")){
            serverSession = serverUser.getCallSession();
        } else if (deviceType.equals("3")){
            serverSession = serverUser.getVoiceSession();
        }
        if (serverSession == null) {
            return;
        }
        RLock lock = redissonClient.getLock(MyConstant.SessionLock + serverSession.getId());
        try {
            if (serverSession.isOpen()){
                boolean locked = lock.tryLock(3, 10, TimeUnit.SECONDS);
                if (locked){
                    if (serverSession.isOpen()){
                        String messageInfo = AESEncrypt.encrypt(message);
                        serverSession.sendMessage(new TextMessage(messageInfo));
                        log.info("发送消息成功messageInfo={},deviceType={},account={}", messageInfo, deviceType,serverUser.getAccount());
                    }
                }
            }
        } catch (Exception e) {
            log.info("发送消息失败，deviceType={},account={}", deviceType,serverUser.getAccount());
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }


    }

}